package com.mzj.eagle.jobs.heartbeat.receiver;

import java.io.IOException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import com.mzj.eagle.jobs.heartbeat.service.HeartbeatService;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.AMQP.BasicProperties;

/**
 * MQ消息监控处理
 */
@Service("heartbeatReceiver")
public class HeartbeatReceiver implements ChannelAwareMessageListener {
	private static final Logger LOG = LoggerFactory.getLogger(HeartbeatReceiver.class);

	@Value("${scale.heartbeat.queue}")
	private String queueName;
	@Value("${scale.heartbeat.exchange}")
	private String exchangeName;
	@Value("${scale.heartbeat.routing.key}")
	private String routingKey;
	
	@Autowired
	private HeartbeatService heartbeatService;

	@Override
	public void onMessage(Message received, Channel channel) {
		LOG.info("---->Start working Machine HeartBeat message...");
		
		if(LOG.isDebugEnabled()) {
			LOG.debug("routingKey:"+routingKey+",queueName:"+queueName+",exchangeName:"+exchangeName);
		}
		byte[] body = received.getBody();
		String message = new String(body);

		if(LOG.isDebugEnabled()) {
			LOG.debug("message : {}", message);
		}

		if (StringUtils.isEmpty(message)) {
			LOG.warn("message is null");
			return;
		}

		boolean isSuccess = true;
		try {
			// 处理
			isSuccess = heartbeatService.insertData(message);
		} catch (Exception e) {
			LOG.error(e.getMessage(), e);
			isSuccess = false;
		}
		try {
			// 成功,清理MQ
			if (isSuccess) {
				channel.basicAck(received.getMessageProperties().getDeliveryTag(), true);
			} else {
				// 处理失败：先删除，在放入到最前
				channel.basicAck(received.getMessageProperties().getDeliveryTag(), true);
				BasicProperties basic=new BasicProperties();
				channel.basicPublish(exchangeName, routingKey, basic, message.getBytes());
//				Thread.sleep(1000);
			}
		} catch (IOException e) {
			LOG.error(e.getMessage(), e);
//		} catch (InterruptedException e) {
//			LOG.error(e.getMessage(), e);
		}

		LOG.info("---->end...");
	}
}